package accumulator;

import org.apache.flink.api.common.accumulators.IntCounter;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.configuration.Configuration;

public class Accumulator {
    public static void main(String[] args) throws Exception {
        //    DataSet计算，创建执行环境
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSource<String> source = env.fromElements("a", "b", "c", "d");

        DataSet<Integer> result = source.map(new RichMapFunction<String, Integer>() {
            IntCounter counter = new IntCounter();

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                //注册一个计数器
                this.getRuntimeContext().addAccumulator("testcounter", counter);
            }

            @Override
            public Integer map(String s) {
                //实现计数功能
                this.counter.add(1);
                return 0;
            }
        });
        //result.print();
        result.writeAsText("./b.text");
        int total = env.execute("11").getAccumulatorResult("testcounter");
        System.out.println("toal: " + total);
    }
}
